package org.lzfto.flink.demo.business.flinkTest.flinkObjects.function.process;

import java.math.BigDecimal;

import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.Window;
import org.apache.flink.util.Collector;
import org.lzfto.flink.demo.business.flinkTest.flinkObjects.selector.OrderMixedKey;
import org.lzfto.flink.demo.business.orders.SimpleOrderAgg;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson2.JSON;

/**
 * 汇总订单处理函数(要求输入和输出的数据都是 SimpleOrderAgg类型)
 * 
 * @param <K> 键类型
 * @param <W> 窗口类型
 */
public class AggOrderProcessWindowFunction<W extends Window>
		extends ProcessWindowFunction<SimpleOrderAgg, SimpleOrderAgg, OrderMixedKey, W> {
	private static final long serialVersionUID = 1L;

	private final Logger LOGGER = LoggerFactory.getLogger(getClass());

	@Override
	public void process(OrderMixedKey key,
			ProcessWindowFunction<SimpleOrderAgg, SimpleOrderAgg, OrderMixedKey, W>.Context context,
			Iterable<SimpleOrderAgg> elements, Collector<SimpleOrderAgg> out) throws Exception {
		SimpleOrderAgg agg = new SimpleOrderAgg(key.getZoneName(), key.getSex(), BigDecimal.ZERO, 0);
		LOGGER.info("窗口正在聚集中,开始处理");
		for (SimpleOrderAgg inval : elements) {
			agg.setCount(agg.getCount() + inval.getCount());
			agg.setTotal(agg.getTotal().add(inval.getTotal()));
			LOGGER.info("-----*** inval={}", JSON.toJSONString(inval));
		}
		LOGGER.info("窗口正在聚集中..,key={},agg={}", JSON.toJSONString(key), JSON.toJSONString(agg));
		out.collect(agg);

	}

}
